package writer.es;

import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import util.WPCommonUtil;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

/**
 * 批量消费消息
 */
public class DataBeanEsConsumer implements Runnable{
    List<TraceDataBean> dataBeans = new ArrayList<>();


    RestHighLevelClient client;

    public DataBeanEsConsumer() {
        this.client = EsClientBuilder.getClient();
    }

    @Override
    public void run() {

        // 1、创建批量操作请求
        BulkRequest request = new BulkRequest();
        while(true) {

            Long startTime = System.currentTimeMillis();
            try  {


                /*
                request.add(new DeleteRequest("mess", "_doc", "3"));
                request.add(new UpdateRequest("mess", "_doc", "2")
                        .doc(XContentType.JSON,"other", "test"));
                request.add(new IndexRequest("mess", "_doc", "4")
                        .source(XContentType.JSON,"field", "baz"));
                */

                // 2、可选的设置
                /*
                request.timeout("2m");
                request.setRefreshPolicy("wait_for");
                request.waitForActiveShards(2);
                */


                //3、发送请求
                // 同步请求
                while (WPCommonUtil.isOK(dataBeans.size(), 100, startTime, 2000l)) {
                    try {
                        TraceDataBean dataBean = TaskQueue.getQueue().take();
                        //此处采用map进行构建
                        HashMap<String, Object> hashMap = new HashMap<>();
                        hashMap.put("lng", dataBean.getLng());
                        hashMap.put("lat", dataBean.getLat());
                        hashMap.put("tel", dataBean.getTel());
                        hashMap.put("eventTime", dataBean.getEventTime());
                        //构建索引请求
                        IndexRequest indexRequest = new IndexRequest("trace");
                        request.add(indexRequest.source(hashMap));
                        dataBeans.add(dataBean);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
                //4、处理响应
                if(bulkResponse != null) {
                    for (BulkItemResponse bulkItemResponse : bulkResponse) {
                        DocWriteResponse itemResponse = bulkItemResponse.getResponse();

                        if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX
                                || bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) {
                            IndexResponse indexResponse = (IndexResponse) itemResponse;
                            //TODO 新增成功的处理

                        } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) {
                            UpdateResponse updateResponse = (UpdateResponse) itemResponse;
                            //TODO 修改成功的处理

                        } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) {
                            DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
                            //TODO 删除成功的处理
                        }
                    }
                }
                dataBeans.clear();

                //异步方式发送批量操作请求
                /*
                ActionListener<BulkResponse> listener = new ActionListener<BulkResponse>() {
                    @Override
                    public void onResponse(BulkResponse bulkResponse) {

                    }

                    @Override
                    public void onFailure(Exception e) {

                    }
                };
                client.bulkAsync(request, listener);
                */

            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }
}
